package vh.sphynx.jobengine;

import java.io.PrintStream;
import java.io.Serializable;
import java.lang.Thread.State;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

public class JobExecutionEngine implements Serializable {

	private static final long serialVersionUID = 1L;
	
	protected LinkedBlockingQueue<Job> jobpools;
	protected String name;
	private boolean done;
	private int workerCount = 0;
	
	transient protected Object jobArgs;
	transient protected List<Thread> workers;
	transient private PrintStream logStream;
	
	protected JobExecutionEngine() {
		
	}

	public JobExecutionEngine(String name, Object jobargs) {
		this.name = name;
		jobpools = new LinkedBlockingQueue<Job>();
		workers = new ArrayList<Thread>();
		this.jobArgs = jobargs;
	}
	
	protected void initialize(Object jobArgs) {
		workers = new ArrayList<Thread>();
		this.jobArgs = jobArgs;
		int n = workerCount;
		workerCount = 0;
		for (int i = 0; i < n; i++)
			addWorker();
	}

	public synchronized void addJob(Job job) {
		job.setEngine(this);
		try {
			synchronized(jobpools) {
				jobpools.offer(job, Long.MAX_VALUE, TimeUnit.DAYS); // wait forever
				jobpools.notifyAll();
			}
		} catch (InterruptedException e) {
			getLogStream().printf("[%s] Error while adding job to pool.\n\tmessage=%s\n", e.getClass().getName(), e.getMessage());
		} 
	}

	protected void processJob() {
		try {
			while (!done) {
				Job job = null;
				synchronized(jobpools) {
					if (jobpools.size() == 0) {
						jobpools.notifyAll();
						jobpools.wait();
					}
					if (jobpools.size() > 0) {
						job = jobpools.poll();
						jobpools.notifyAll();
					}
				}
				if (job != null) job.executeJob(jobArgs);
			}
			synchronized(workers) {
				workers.remove(Thread.currentThread());
				workers.notifyAll();
			}
		}
		catch(Exception ex) {
			ex.printStackTrace();
			synchronized(workers) {
				workers.remove(Thread.currentThread());
				workers.notifyAll();
			}
			synchronized(jobpools) {
				jobpools.notifyAll();
			}
		}
	}

	public void waitForDone() {
		try {
			if (done)
				close();
			else {
				while(!done) {
					synchronized(jobpools) {
						boolean allSleep = true;
						for(Thread t: workers)
							if (t.getState() == State.RUNNABLE) allSleep = false;
						if (!done && !allSleep)
							jobpools.wait();
						else if (jobpools.size() == 0 || workers.size() == 0) 
							done = true;
					}
				}
				close();
			}
		} catch (InterruptedException e) {
			close();
		}
	}

	public void close() {
		done = true;
		synchronized(jobpools) {
			jobpools.clear();
			jobpools.notifyAll();
		}
		try {
			while (workers.size() > 0)
				synchronized(workers) {
					if (workers.size() > 0) workers.wait();
				}
			getLogStream().close();
		} catch (Exception e) {
		}
	}

	public void addWorker() {
		String name = this.name + "-" + (++workerCount);
		Thread wt = new Thread(new Runnable() {
			@Override
			public void run() {
				processJob();
			}
		});
		wt.setName(name);
		workers.add(wt);
		wt.start();
	}

	public void setLogStream(PrintStream logStream) {
		this.logStream = logStream;
	}

	public PrintStream getLogStream() {
		return logStream;
	}

}